package org.company.saturn.hadoop.yarn;

import org.apache.commons.lang.math.RandomUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.map.InverseMapper;
import org.apache.hadoop.mapreduce.lib.map.RegexMapper;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.reduce.LongSumReducer;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.sun.jersey.core.impl.provider.entity.XMLJAXBElementProvider.Text;

public class Grep extends Configured implements Tool {
	
	private static Logger logger = LoggerFactory.getLogger(Grep.class);

	@Override
	public int run(String[] args) throws Exception {
		if(args.length < 3) {
			logger.error("Grep <inDir> <outDir> <regex> [<group>]");
			ToolRunner.printGenericCommandUsage(System.err);
			return 2;
		}
		
		Path tmpPath = new Path("grep-temp-" + RandomUtils.nextInt());
		
		Configuration conf = this.getConf();
		conf.set(RegexMapper.PATTERN, args[2]);
		if(args.length == 4)
			conf.set(RegexMapper.PATTERN, args[3]);
		
		Job grepJob = new Job(conf);
		grepJob.setJobName("grep-search");
		FileInputFormat.setInputPaths(grepJob,args[0]);
		
		grepJob.setMapperClass(RegexMapper.class);
		grepJob.setCombinerClass(LongSumReducer.class);
		grepJob.setReducerClass(LongSumReducer.class);
		
		FileOutputFormat.setOutputPath(grepJob, tmpPath);
		grepJob.setOutputFormatClass(SequenceFileOutputFormat.class);
		grepJob.setOutputKeyClass(Text.class);
		grepJob.setOutputValueClass(LongWritable.class);
		
		grepJob.waitForCompletion(true);
		
		Job sortJob = new Job(conf);
		sortJob.setJobName("grep-sort");
		FileInputFormat.setInputPaths(sortJob, tmpPath);
		sortJob.setInputFormatClass(SequenceFileInputFormat.class);
		
		sortJob.setMapperClass(InverseMapper.class);
		sortJob.setNumReduceTasks(1);
		
		FileOutputFormat.setOutputPath(sortJob, new Path(args[1]));
		sortJob.setSortComparatorClass(LongWritable.DecreasingComparator.class);
		sortJob.waitForCompletion(true);
		
		return 0;
	}

}
